package com.alibaba.otter.node.etl.load.loader.mq.rabbit.strategy;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.MqPlugin;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.Publisher;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqConsumeException;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.jboss.netty.util.internal.StringUtil;

@Slf4j
@NoArgsConstructor
public class SendToDLQ implements RescueAction {

    public SendToDLQ(String exchange, String routingKey) {
        this.namespace = null;
        this.exchange = exchange;
        this.routingKey = routingKey;
    }

    public SendToDLQ(String namespace, String exchange, String routingKey) {
        this.namespace = namespace;
        this.exchange = exchange;
        this.routingKey = routingKey;
    }

    private String namespace;
    private String exchange;
    private String routingKey;

    private Publisher publisher;

    @Override
    public int handle(MqConsumeException ex) {
        try {
            if (publisher == null) {
                exchange = exchange == null ? "" : exchange;
                routingKey = StringUtils.isEmpty(routingKey) ? ex.getMsgCtx().getQueue() + ".dlq" : routingKey;
                String namespace = this.namespace != null ? this.namespace : ex.getMsgCtx().getNamespace();
                publisher = MqPlugin.init().publisher(namespace).toExchange(exchange).routingKey(routingKey).build();
            }
            publisher.send(ex.getMessage());
        } catch (Exception e) {
            log.error("Failed to send Message to DLQ.", e);
        }
        return 1;
    }
}
